package com.fulihui.wgj.consumer.listener;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fulihui.common.constants.TopicConstants;
import com.fulihui.common.enums.SendLogStatusEnum;
import com.fulihui.common.util.CloneUtils;
import com.fulihui.common.util.JedisUtil;
import com.fulihui.common.util.TemplateDisabledUtil;
import com.fulihui.common.util.TemplateMsgCancelUtil;
import com.fulihui.open.shortlink.model.WgjSendLink;
import com.fulihui.weixinsdk.model.TemplateItem;
import com.fulihui.wgj.consumer.config.property.ShortLinkProperties;
import com.fulihui.wgj.consumer.result.BatchSendResult;
import com.fulihui.wgj.consumer.service.TemplateMsgService;
import com.fulihui.wgj.consumer.service.WxMsgService;
import com.fulihui.wgj.facade.dto.MsgUserDTO;
import com.fulihui.wgj.facade.dto.TemplateMsgDTO;
import com.fulihui.wgj.facade.msg.TemplateMsg;
import com.fulihui.wgj.facade.request.template.TemplateLogSendUpdateRequest;
import com.fulihui.wgj.facade.service.template.SendTemplateLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.DigestUtils;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.Date;
import java.util.List;
import java.util.Map;

import static com.fulihui.common.constants.PlaceholderConstants.NICK_NAME;

/**
 * @author: LiuJie
 * @version: 2018/7/26 17:26
 * @description:
 */
@Slf4j
@Component
public class TemplateMsgListener extends AbstractListener<TemplateMsg> {

    @Autowired
    private TemplateMsgService templateMsgService;
    @Reference(version = "1.0.0")
    private SendTemplateLogService sendTemplateLogService;


    @Override
    @KafkaListener(topics = TopicConstants.TEMPLATE_MSG)
    void listener(ConsumerRecord<String, String> record) {

        if (null == record || null == record.value()) {
            log.error("kafka 消息为空，record={}", record);
            return;
        }
        TemplateMsg msg = super.unpack(record.value(), TemplateMsg.class);
        execute(msg);
    }

    /**
     * 执行具体业务
     *
     * @param msg 具体消息通配
     */
    @Override
    void execute(TemplateMsg msg) {
        super.execute(msg);
        BatchSendResult result = templateMsgService.batchHandle(msg);
        if (result == null) {
            return;
        }

        //  成功失败数 写入数据库 进行累加
        recordSendResult(msg, result);
    }



    /**
     * 累加发送结果
     * @param msg
     * @param result
     */
    private void recordSendResult(TemplateMsg msg, BatchSendResult result) {
        Long recordId = msg.getRecordId();
        TemplateLogSendUpdateRequest request = new TemplateLogSendUpdateRequest();
        request.setId(recordId);
        request.setSuccessTotal(result.getSuccessNum());
        request.setFailTotal(result.getFailNum());
        if (msg.getIsLast()) {
            request.setSendEndTime(new Date());
            request.setSendStatus(SendLogStatusEnum.complete.getCode());
        }
        sendTemplateLogService.updateSendRecordAccumulate(request);
    }



}
